package com.lzcu.maoyan.mapreduce.mapreduce;

import com.lzcu.maoyan.mapreduce.mapreduce.mapper.MovieLeadingActorMapper;
import com.lzcu.maoyan.mapreduce.mapreduce.reduce.MovieLeadingActorReduce;
import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieLeadingActorWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.phoenix.mapreduce.PhoenixInputFormat;
import org.apache.phoenix.mapreduce.PhoenixOutputFormat;
import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;

/**
 * 演员主角
 * @author WangLiang
 * @date `2024/4/27`
 */
public class MovieLeadingActorMR extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {

        //1.创建job
        Configuration conf = new Configuration();
        conf.addResource("phoenix/core-site.xml");
        conf.addResource("phoenix/hbase-site.xml");
        conf.addResource("phoenix/hdfs-site.xml");
        Job job = Job.getInstance(conf, "PhoenixStockApplication");
        job.setJarByClass(PhoenixMR.class);
        //2.设置输入输出格式
        job.setInputFormatClass(PhoenixInputFormat.class);
        job.setOutputFormatClass(PhoenixOutputFormat.class);
        //3.设置数据读入和写出路径
        String selectQuery = "SELECT\n" +
                "    M1.MOVIE_NAME_INFO AS MOVIE_NAME,\n" +
                "    M1.MAX_TOTAL_SESSIONS AS MAX_TOTAL_SESSIONS,\n" +
                "    M2.ACTTHELEADINGROLE AS ACTTHELEADINGROLE\n" +
                "FROM\n" +
                "    (SELECT MOVIE_NAME_INFO, MAX(TOTAL_SESSIONS) AS MAX_TOTAL_SESSIONS\n" +
                "     FROM MAOYANMOVIE.MOVIE_INFO\n" +
                "     GROUP BY MOVIE_NAME_INFO) M1\n" +
                "JOIN\n" +
                "     MAOYANMOVIE.MOVIE_DETAILED M2\n" +
                "ON\n" +
                "    M1.MOVIE_NAME_INFO = M2.MOVIENAME AND YEAR(TO_DATE(M2.RELEASEDATE)) = 2024";
        PhoenixMapReduceUtil.setInput(job, MovieLeadingActorWritable.class,
                "MAOYANMOVIE.MOVIE_INFO,MAOYANMOVIE.MOVIE_DETAILED", selectQuery);
        PhoenixMapReduceUtil.setOutput(job,
                "MAOYANMOVIE.movie_LeadingActor", "LEADING_ACTOR,BOX_OFFICE,MOVIE,MOVIECOUNT");
        //4.设置代码片段
        job.setMapperClass(MovieLeadingActorMapper.class);
        job.setReducerClass(MovieLeadingActorReduce.class);
        //5.设置Mapper和Reducer端的输出key,value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(MovieLeadingActorWritable.class);
        TableMapReduceUtil.addDependencyJars(job);
        //6.任务提交
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new MovieLeadingActorMR(), args);
    }

}
